package net.xuele.learn.rsocket;

import io.rsocket.Payload;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;

/**
 * @author: patrick-Hikari
 * @description:
 * @date: 2021/8/19 10:00 上午
 */
public class ReqStreamClient {

    private final RSocket socket;

    public ReqStreamClient() {
        this.socket = RSocketFactory.connect()
                .transport(TcpClientTransport.create("localhost", 8070))
                .start()
                .block();
    }

    public Flux<Float> getDataStream() {
        return socket
                /*发送请求*/
                .requestStream(DefaultPayload.create("data"))
                /* 尝试得到了代表结果的Flux */
                .map(Payload::getData)
                .map(buf -> buf.getFloat())
                .onErrorReturn(null);
    }

    public void dispose() {
        this.socket.dispose();
    }
}
